

#include <vector>
#include <string>
#include <sstream>

#include <boost/algorithm/string.hpp>


#include "bocat/common/logging.hpp"

#include "boque/trasit/channel.hpp"

namespace boque		{
namespace trasit	{


///////////////////////////////////////////////////////////////////////
//

Channel::Channel( const std::string& address, const std::string& buf_dir )
	:	_address( address ),
		_buf_dir( buf_dir ),
		_buffesq( _buf_dir ),
		_broker_p( NULL ),
		_status( UNKNOWN ),
		_waiter_p( NULL )
{
	std::string	protocol, host, port;

	// address format : tcp://127.0.0.1:5555
	parseAddress( address, protocol, host, port );

	_broker_p = new Broker( host, port );
	_broker_connected = false;

	if ( !_buffesq.empty() ) {
		// if queue is not empty, channel is buffered
		_status = BUFFERED;

		bocat::common::Logger*	logger_p = bocat::common::getLogger();
		logger_p->error(__LINE__, __FILE__, "channel to %s enter BUFFER mode", _address.c_str());
	}

	_relay_enable = false;
	_relaying = false;
}

Channel::~Channel()
{
	if ( NULL != _waiter_p ) {
		delete _waiter_p;
		_waiter_p = NULL;
	}

	if ( NULL != _broker_p ) {
		delete _broker_p;
		_broker_p = NULL;
	}
}


///////////////////////////////////////////////////////////////////////
//

int
Channel::send( const std::string& data )
{
	return	send( data.c_str(), data.size() );
}

int
Channel::send( const char* data, size_t size )
{
	boost::mutex::scoped_lock	_scoped_lock( _mutex );

	if ( BUFFERED == _status ) {
		// push data to disk queue
		_buffesq.push( data, size );
	}
	else {
		// try to send data directly
		try {
			int	rc = _broker_p->send( data, size );
			if ( rc < 0 ) {
				bocat::common::Logger*	logger_p = bocat::common::getLogger();
				logger_p->error(__LINE__, __FILE__, "fail to send data to %s ", _address.c_str());
			}
		}
		catch( std::exception& e ) {
			bocat::common::Logger*	logger_p = bocat::common::getLogger();
			logger_p->error(__LINE__, __FILE__, "fail to send data to %s -- %s", _address.c_str(), e.what());
			logger_p->error(__LINE__, __FILE__, "channel to %s enter BUFFER mode", _address.c_str());

			_status = BUFFERED;
			_broker_connected = false;

			// push data to disk queue
			_buffesq.push( data, size );
		}
	}


	return	0;
}

///////////////////////////////////////////////////////////////////////
//

static	uint32_t
getSleepTime( uint32_t retry_times )
{
	uint32_t	seconds = 15;

	if ( retry_times < 10 ) {
		seconds = 1;
	}
	else if ( retry_times < 20 ) {
		seconds = 3;
	}
	else if ( retry_times < 50 ) {
		seconds = 5;
	}
	else if ( retry_times < 100 ) {
		seconds = 10;
	}

	return	(1000 * seconds);
}


int
Channel::relay( void )
{
	if ( !_relay_enable ) {
		return	0;
	}

	boost::mutex::scoped_lock	_scoped_lock( _mutex_relay );

    bocat::common::Logger*	logger_p = bocat::common::getLogger();

	uint32_t	retry_times = 0;
	while ( _relay_enable ) {
		if ( ALIVE == _status ) {
			boost::this_thread::sleep(boost::posix_time::milliseconds( 1000 ));
			continue;
		}

		if ( false == _broker_connected ) {
			// try to connect 
			try {
				_broker_p->connect();
				_broker_connected = true;
			}
			catch (std::exception& e) {
				bocat::common::Logger*	logger_p = bocat::common::getLogger();
				logger_p->error(__LINE__, __FILE__, "fail to connect %s -- %s", _address.c_str(), e.what());
				if ( BUFFERED != _status ) {
					logger_p->error(__LINE__, __FILE__, "channel to %s enter BUFFER mode", _address.c_str());
					_status = BUFFERED;
				}

				_broker_connected = false;
			}
		}
		if ( false == _broker_connected ) {
			uint32_t	ms = getSleepTime( ++retry_times );
			boost::this_thread::sleep(boost::posix_time::milliseconds( ms ));

			continue;
			//return	0;
		}
		retry_times = 0;

		logger_p->info(__LINE__, __FILE__, "relay to %s ...", _address.c_str());

		// send data in queue
		int	rc = do_relay();
		if ( rc < 0 ) {
			continue;
			//return	-1;
		}

		// send done
		// try to set status as alive
		{
			boost::mutex::scoped_lock	_scoped_lock( _mutex );

			// TODO : to clear queue
			int	rc = do_relay();
			if ( rc < 0 ) {
				continue;
				//return	-1;
			}

			if ( ALIVE != _status ) {
				_status = ALIVE;

				//bocat::common::Logger*	logger_p = bocat::common::getLogger();
				logger_p->error(__LINE__, __FILE__, "channel to %s enter ALIVE mode", _address.c_str());
			}
		}

		//break;
	}	//// while ( _relay_enable )

    logger_p->info(__LINE__, __FILE__, "relay to %s done.", _address.c_str());
    
	return	0;
}


int
Channel::do_relay( void )
{
    _relaying = true;
    
	while ( !_buffesq.empty() ) {
		if ( !_relay_enable ) {
            _relaying = false;
			return	-2;
		}

		std::string	data = _buffesq.front();

		// try to send data directly
		try {
			_broker_p->send( data );
		}
		catch( std::exception& e ) {
			//bocat::common::Logger*	logger_p = bocat::common::getLogger();
			//logger_p->error(__LINE__, __FILE__, "fail to send data to %s -- %s", _address.c_str(), e.what());
			//logger_p->error(__LINE__, __FILE__, "channel to %s enter BUFFER mode", _address.c_str());

			_status = BUFFERED;
			_broker_connected = false;

            _relaying = false;
			return	-1;
		}

		// just discard pop result
		data = _buffesq.pop();
	}

    _relaying = false;
	return	0;
}


///////////////////////////////////////////////////////////////////////
//

void
Channel::parseAddress( const std::string& address, std::string& protocol, std::string& host, std::string& port )
{
	std::string::size_type	pos = address.find( "://" );
	if ( std::string::npos == pos ) {
		std::ostringstream	msg;
		msg << "Bad address format : '" << address << "'";

		throw std::runtime_error( msg.str() );
	}
	protocol = address.substr( 0, pos );
	std::string	host_port = address.substr( pos + 3 );

	std::vector< std::string >	b_v;
	boost::split( b_v, host_port, boost::is_any_of(":") );
	if ( 2 != b_v.size() ) {
		std::ostringstream	msg;
		msg << "Bad address format : '" << address << "'";

		throw std::runtime_error( msg.str() );
	}

	host = b_v[ 0 ];
	port = b_v[ 1 ];

	boost::trim( protocol );
	boost::trim( host );
	boost::trim( port );
}


}	//// namespace trasit
}	//// namespace boque

